package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.uitls.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        //connection.setAutoCommit(true);
    }

    //value:{"database":"gmall-210826-flink","tableName":"base_trademark","before":{},"after":{"id":"...","tm_name":"...","logo_url":"..."},"type":"insert","sinkTable":"dim_base_trademark"}
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        PreparedStatement preparedStatement = null;

        try {
            //1.构建SQL语句
            String sql = genSql(value.getString("sinkTable"), value.getJSONObject("after"));
            System.out.println(sql);

            //2.编译SQL
            preparedStatement = connection.prepareStatement(sql);

            //如果为更新数据，则先删除Redis中的数据
            if ("update".equals(value.getString("type"))) {
                DimUtil.deleteRedisDimInfo(value.getString("sinkTable").toUpperCase(),
                        value.getJSONObject("after").getString("id"));
            }

            //3.执行写入数据操作
            preparedStatement.execute();
            connection.commit();

        } catch (SQLException e) {
            System.out.println("插入数据失败！");
            e.printStackTrace();
        } finally {
            //4.释放资源
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }


    }

    //构建SQL语句: upsert into db.tn(aa,bb,cc) values('..','..','..')
    //after:{"aa":"","bb":"","cc":""}
    private String genSql(String tableName, JSONObject after) {

        //scala : list.mkString(",")
        Set<String> columns = after.keySet();
        Collection<Object> values = after.values();

        return "upsert into " + GmallConfig.HBASE_SCHEMA + "." + tableName + "(" +
                StringUtils.join(columns, ",") + ") values('" +
                StringUtils.join(values, "','") + "')";
    }
}
